/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.store.ha;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.store.SelectMappedBufferResult;

/**
 * 主服务端 HA 连接对象的封装，当主服务器接收到从服务器发过来的消息后，会封装成一个 HAConnection 对象，
 * 其中里面又封装了读 Socket 连接实现与 写 Socket 连接实现：
 * HAConnection$ReadSocketService：主服务器读实现类
 * HAConnection$WriteSocketService：主服务器写实现类
 */
public class HAConnection {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    //主从同步的核心实现类本身
    private final HAService haService;
    //从服务器的连接过来，主服务器接收到的socketChannel对象
    private final SocketChannel socketChannel;
    //从服务器的IP地址
    private final String clientAddr;
    //主服务器写实现类
    private WriteSocketService writeSocketService;
    //主服务器读实现类
    private ReadSocketService readSocketService;

    //从服务器传过来的请求偏移量
    private volatile long slaveRequestOffset = -1;
    //从客户端传过来的确认偏移量
    private volatile long slaveAckOffset = -1;

    public HAConnection(final HAService haService, final SocketChannel socketChannel) throws IOException {
        this.haService = haService;
        this.socketChannel = socketChannel;
        //通过socketChannel对象获取从服务器的客户端的地址
        this.clientAddr = this.socketChannel.socket().getRemoteSocketAddress().toString();
        this.socketChannel.configureBlocking(false);
        /**
         * setSoLinger(true, delay_time)
         *
         * 当网卡收到关闭连接请求后，等待delay_time
         *
         * 如果在delay_time过程中数据发送完毕，正常四次挥手关闭连接
         *
         * 如果在delay_time过程中数据没有发送完毕，发送RST包关闭连接
         */
        this.socketChannel.socket().setSoLinger(false, -1);
        this.socketChannel.socket().setTcpNoDelay(true);
        //设置写缓冲区大小64k
        this.socketChannel.socket().setReceiveBufferSize(1024 * 64);
        //设置读缓冲区大小64k
        this.socketChannel.socket().setSendBufferSize(1024 * 64);
        this.writeSocketService = new WriteSocketService(this.socketChannel);
        this.readSocketService = new ReadSocketService(this.socketChannel);
        //把连接数+1
        this.haService.getConnectionCount().incrementAndGet();
    }

    public void start() {
        this.readSocketService.start();
        this.writeSocketService.start();
    }

    public void shutdown() {
        this.writeSocketService.shutdown(true);
        this.readSocketService.shutdown(true);
        this.close();
    }

    public void close() {
        if (this.socketChannel != null) {
            try {
                this.socketChannel.close();
            } catch (IOException e) {
                HAConnection.log.error("", e);
            }
        }
    }

    public SocketChannel getSocketChannel() {
        return socketChannel;
    }

    class ReadSocketService extends ServiceThread {
        //读的缓冲区最大大小 1m
        private static final int READ_MAX_BUFFER_SIZE = 1024 * 1024;
        private final Selector selector;
        private final SocketChannel socketChannel;
        //创建一个ByteBuffer
        private final ByteBuffer byteBufferRead = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);
        //处理的位置
        private int processPosition = 0;
        //上次读取时间。指的是从服务器客户端上报偏移量的读取
        private volatile long lastReadTimestamp = System.currentTimeMillis();

        public ReadSocketService(final SocketChannel socketChannel) throws IOException {
            this.selector = RemotingUtil.openSelector();
            this.socketChannel = socketChannel;
            this.socketChannel.register(this.selector, SelectionKey.OP_READ);
            //守护线程
            this.setDaemon(true);
        }

        @Override
        public void run() {
            HAConnection.log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    this.selector.select(1000);
                    //处理从服务器上报偏移量的请求
                    boolean ok = this.processReadEvent();
                    if (!ok) {
                        HAConnection.log.error("processReadEvent error");
                        break;
                    }

                    //当前时间减去上次的读取时间
                    long interval = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now() - this.lastReadTimestamp;
                    //如果超时了，退出循环
                    if (interval > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaHousekeepingInterval()) {
                        log.warn("ha housekeeping, found this connection[" + HAConnection.this.clientAddr + "] expired, " + interval);
                        break;
                    }
                } catch (Exception e) {
                    HAConnection.log.error(this.getServiceName() + " service has exception.", e);
                    break;
                }
            }

            //让自己停止
            this.makeStop();
            //让写socket服务线程也停止
            writeSocketService.makeStop();

            //把这个连接从集合中移除
            haService.removeConnection(HAConnection.this);

            //给连接数-1
            HAConnection.this.haService.getConnectionCount().decrementAndGet();
            //意思是将socketChannel注册到这个this.selector上，如果已经注册过了。。直接返回对应的SelectionKey
            SelectionKey sk = this.socketChannel.keyFor(this.selector);
            if (sk != null) {
                sk.cancel();
            }

            try {
                this.selector.close();
                this.socketChannel.close();
            } catch (IOException e) {
                HAConnection.log.error("", e);
            }

            HAConnection.log.info(this.getServiceName() + " service end");
        }

        @Override
        public String getServiceName() {
            return ReadSocketService.class.getSimpleName();
        }

        /**
         * 处理从服务器上报偏移量的请求
         * @return
         *
         * 获取从服务器上报的偏移量；
         * 唤醒主从同步消费者发送消息同步返回的线程，该方法实现了主从同步-同步复制的功能。
         */
        private boolean processReadEvent() {
            int readSizeZeroTimes = 0;

            //如果当前的读缓冲区中没有字节可读
            if (!this.byteBufferRead.hasRemaining()) {
                //则把读缓冲区变为可写模式
                this.byteBufferRead.flip();
                //处理的位置也标识为0
                this.processPosition = 0;
            }

            //如果读缓冲区中有字节可读，循环读取
            while (this.byteBufferRead.hasRemaining()) {
                try {
                    //从socketChannel里面读出内容放进byteBufferRead里面去
                    int readSize = this.socketChannel.read(this.byteBufferRead);
                    if (readSize > 0) {
                        readSizeZeroTimes = 0;
                        //把当前时间赋值给上次读取时间
                        this.lastReadTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
                        //表示当前传进来的byteBufferRead.position()减去上次的处理过的位置大于8
                        //processPosition默认是0
                        if ((this.byteBufferRead.position() - this.processPosition) >= 8) {
                            //通过这个拿到这次应该从byteBufferRead的哪个位置开始读取数据
                            int pos = this.byteBufferRead.position() - (this.byteBufferRead.position() % 8);
                            //从缓冲区里面拿出这次从客户端传过来的偏移量
                            long readOffset = this.byteBufferRead.getLong(pos - 8);
                            //记录这次循环应该从byteBufferRead的哪个位置读取数据
                            this.processPosition = pos;
                            //把偏移量赋值给slaveAckOffset
                            HAConnection.this.slaveAckOffset = readOffset;
                            if (HAConnection.this.slaveRequestOffset < 0) {
                                HAConnection.this.slaveRequestOffset = readOffset;
                                log.info("slave[" + HAConnection.this.clientAddr + "] request offset " + readOffset);
                            }

                            //唤醒主从同步消费者发送消息同步返回的线程，该方法实现了主从同步-同步复制的功能。
                            HAConnection.this.haService.notifyTransferSome(HAConnection.this.slaveAckOffset);
                        }
                    } else if (readSize == 0) {
                        //记录从从服务器读取请求偏移量的次数，如果超过或等于3次。退出循环
                        if (++readSizeZeroTimes >= 3) {
                            break;
                        }
                    } else {
                        log.error("read socket[" + HAConnection.this.clientAddr + "] < 0");
                        return false;
                    }
                } catch (IOException e) {
                    log.error("processReadEvent exception", e);
                    return false;
                }
            }

            return true;
        }
    }

    /**
     * 计算需要拉取的偏移量，如果从服务器第一次拉取，则从最后一个 commitLog 文件的初始偏移量开始同步；
     * 传输消息到从服务器；
     * 发送心跳包到从服务器，保持长连接。
     */
    class WriteSocketService extends ServiceThread {
        private final Selector selector;
        private final SocketChannel socketChannel;
        //消息头的大小
        private final int headerSize = 8 + 4;
        //新建一个消息头的缓冲区
        private final ByteBuffer byteBufferHeader = ByteBuffer.allocate(headerSize);
        //下次传输的位置
        private long nextTransferFromWhere = -1;
        private SelectMappedBufferResult selectMappedBufferResult;
        //上次传输是否成功的标志
        private boolean lastWriteOver = true;
        //上次传输的时间
        private long lastWriteTimestamp = System.currentTimeMillis();

        public WriteSocketService(final SocketChannel socketChannel) throws IOException {
            this.selector = RemotingUtil.openSelector();
            this.socketChannel = socketChannel;
            this.socketChannel.register(this.selector, SelectionKey.OP_WRITE);
            this.setDaemon(true);
        }

        @Override
        public void run() {
            HAConnection.log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                    this.selector.select(1000);

                    //这个this.slaveRequestOffset是这个类的223行赋值的。也就是客户端上报的消息物理偏移量
                    //如果slaveRequestOffset=-1，说明读线程还没有获取从服务器的偏移量，继续循环等待
                    if (-1 == HAConnection.this.slaveRequestOffset) {
                        Thread.sleep(10);
                        continue;
                    }

                    // 如果nextTransferFromWhere=-1，说明线程刚开始执行数据传输，之前就没有同步过消息到从服务器
                    if (-1 == this.nextTransferFromWhere) {
                        // 如果slaveRequestOffset=0，说明从服务器是第一次上报偏移量，表示备用Broker端还没有commitlog数据
                        if (0 == HAConnection.this.slaveRequestOffset) {
                            // 获取最后一个commitlog文件最大的物理偏移量
                            long masterOffset = HAConnection.this.haService.getDefaultMessageStore().getCommitLog().getMaxOffset();
                            // 求出下次需要传输的消息的物理偏移量
                            masterOffset =
                                masterOffset
                                    - (masterOffset % HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig()
                                    .getMappedFileSizeCommitLog());

                            if (masterOffset < 0) {
                                masterOffset = 0;
                            }
                            this.nextTransferFromWhere = masterOffset;
                        } else {
                            //如果slaveRequestOffset!=0,说明不是第一次接收到偏移量，
                            // 则把当前服务端接收的偏移量赋值给下一次需要传输的偏移量
                            this.nextTransferFromWhere = HAConnection.this.slaveRequestOffset;
                        }

                        log.info("master transfer data from " + this.nextTransferFromWhere + " to slave[" + HAConnection.this.clientAddr
                            + "], and slave request " + HAConnection.this.slaveRequestOffset);
                    }
                    // 判断上次写事件是否已全部写完成
                    if (this.lastWriteOver) {
                        //计算是否已到发送心跳包时间
                        long interval =
                            HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp;

                        // 发送心跳包，以保持长连接，说明已经超过5秒钟
                        if (interval > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig()
                            .getHaSendHeartbeatInterval()) {

                            // Build Header

                            //向备用Broker发送心跳消息，消息为12个字节，前8个字节为开始同步的偏移量offset，
                            // 后4个字节填0；若发送成功则继续下面的逻辑，否则从第1步开始重新执行；
                            this.byteBufferHeader.position(0);
                            this.byteBufferHeader.limit(headerSize);
                            this.byteBufferHeader.putLong(this.nextTransferFromWhere);
                            this.byteBufferHeader.putInt(0);
                            this.byteBufferHeader.flip();

                            //发送心跳包给从服务器
                            this.lastWriteOver = this.transferData();
                            if (!this.lastWriteOver)
                                //发送失败，从while头开始再循环
                                continue;
                        }
                    } else {
                        //表示上一次没有传输完。。先把上一次的传输完。。再传这次的
                        this.lastWriteOver = this.transferData();
                        if (!this.lastWriteOver)
                            //发送失败，从while头开始再循环
                            continue;
                    }

                   // 以nextTransferFromWhere为开始读取偏移量从commitlog中读取数据，
                    // 调用DefaultMessageStore对象的getCommitLogData方法；
                    SelectMappedBufferResult selectResult =
                        HAConnection.this.haService.getDefaultMessageStore().getCommitLogData(this.nextTransferFromWhere);
                    if (selectResult != null) {
                        int size = selectResult.getSize();

                      //  再检查该数据的大小是否大于了32K，每次数据同步最多只能同步32K，若大于了32K，则只发送前32K数据；
                        //如果这次需要传输的数据大于配置的批量传输数据量
                        if (size > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize()) {
                            //则把配置的批量传输数据量赋值给这次的传输数量
                            size = HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize();
                        }

                        /**
                         * 消息结构为：
                         *
                         * 12个字节的消息头，其中，前8个字节为开始同步的偏移量offset，
                         * 后4个字节为同步数据的大小；先发送消息头，发送完成之后再发送同步数据；
                         */

                        long thisOffset = this.nextTransferFromWhere;
                        //把这次传输的数据量大小+下次传输的位置。做为下次传输的起始位置
                        this.nextTransferFromWhere += size;

                        selectResult.getByteBuffer().limit(size);
                        this.selectMappedBufferResult = selectResult;

                        // Build Header
                        this.byteBufferHeader.position(0);
                        this.byteBufferHeader.limit(headerSize);
                        this.byteBufferHeader.putLong(thisOffset);
                        this.byteBufferHeader.putInt(size);
                        this.byteBufferHeader.flip();
                        //消息头传输到对应的从节点
                        this.lastWriteOver = this.transferData();
                    } else {
                       // 若没有获取到数据则该服务线程等待100毫秒之后重新从while头开始再循环
                        HAConnection.this.haService.getWaitNotifyObject().allWaitForRunning(100);
                    }
                } catch (Exception e) {

                    HAConnection.log.error(this.getServiceName() + " service has exception.", e);
                    break;
                }
            }

            //当写线程停止运行的时候，把在389行之前加进去的这个线程给去掉
            HAConnection.this.haService.getWaitNotifyObject().removeFromWaitingThreadTable();

            if (this.selectMappedBufferResult != null) {
                this.selectMappedBufferResult.release();
            }

            //写线程停止
            this.makeStop();

            //读线程停止
            readSocketService.makeStop();

            haService.removeConnection(HAConnection.this);

            SelectionKey sk = this.socketChannel.keyFor(this.selector);
            if (sk != null) {
                sk.cancel();
            }

            try {
                this.selector.close();
                this.socketChannel.close();
            } catch (IOException e) {
                HAConnection.log.error("", e);
            }

            HAConnection.log.info(this.getServiceName() + " service end");
        }

        private boolean transferData() throws Exception {
            int writeSizeZeroTimes = 0;
            // Write Header
            while (this.byteBufferHeader.hasRemaining()) {
                //将消息头传输给从服务器
                int writeSize = this.socketChannel.write(this.byteBufferHeader);
                if (writeSize > 0) {
                    writeSizeZeroTimes = 0;
                    //更新上次写入时间
                    this.lastWriteTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
                } else if (writeSize == 0) {
                    if (++writeSizeZeroTimes >= 3) {
                        break;
                    }
                } else {
                    throw new Exception("ha master write header error < 0");
                }
            }

            //selectMappedBufferResult这个里面装的是根据上报的消息偏移量来在commitlog文件查找出来的消息
            if (null == this.selectMappedBufferResult) {
                return !this.byteBufferHeader.hasRemaining();
            }

            writeSizeZeroTimes = 0;

            // Write Body
            if (!this.byteBufferHeader.hasRemaining()) {
                while (this.selectMappedBufferResult.getByteBuffer().hasRemaining()) {
                    int writeSize = this.socketChannel.write(this.selectMappedBufferResult.getByteBuffer());
                    if (writeSize > 0) {
                        writeSizeZeroTimes = 0;
                        this.lastWriteTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
                    } else if (writeSize == 0) {
                        if (++writeSizeZeroTimes >= 3) {
                            break;
                        }
                    } else {
                        throw new Exception("ha master write body error < 0");
                    }
                }
            }

            boolean result = !this.byteBufferHeader.hasRemaining() && !this.selectMappedBufferResult.getByteBuffer().hasRemaining();

            if (!this.selectMappedBufferResult.getByteBuffer().hasRemaining()) {
                this.selectMappedBufferResult.release();
                this.selectMappedBufferResult = null;
            }

            return result;
        }

        @Override
        public String getServiceName() {
            return WriteSocketService.class.getSimpleName();
        }

        @Override
        public void shutdown() {
            super.shutdown();
        }
    }
}
